package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.message.ArticleVisitStreamMess;
import com.heima.model.message.UpdateArticleMess;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * 热点文章-流式处理的类
 */
@Component
public class HotArticleStream {

    @Bean
    public KStream<String,String > kStream(StreamsBuilder streamsBuilder){
        //1.接收消息
       KStream<String ,String > kStream= streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        //2.流式处理
        kStream.map(new KeyValueMapper<String, String, KeyValue<String, String >>() {
            /**
             *定义
             * @param key --null
             * @param value--- json类型字符串的消息内容 updateArticleMess
             * @return  key=文章id
             *          value=${type}:1  ==比如此刻是阅读行为， VIEWS:1
             */
            @Override
            public KeyValue<String , String > apply(String key, String value) {
                //1.接收消息，并转化格式
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                //2.返回是key=文章id, value=${type}:${add}
                return new KeyValue<String ,String >(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
            }
        })
                /**
                 * key=文章id
                 *  value=${type}:${add}
                 */
                .groupBy((key,value)-> key)//以文章id进行分组
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//多长时间聚合一次
        /**
         *聚合处理
         * 我们要达到的目标是： articleId 对应views数量，collection数量，comment数量，likes数量
         */
                .aggregate(
                        /**
                         * 初始化
                         * 先定义COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                         */
                        new Initializer<String>() {
                               @Override
                               public String apply() {
                                   return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                               }
                           },
                        /**
                         * 开始执行聚合操作
                         * 参数解析
                         * String key, 此刻表示的是文章id
                         * String value,  此刻表示${type}:${add}
                         * String init 此刻表示初始化的返回值COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                         */
                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String init) {
                                //1.判断value
                                if(StringUtils.isBlank(value)){
                                    return init;
                                }
                                //初始化值
                                Integer collection = 0,comment=0,like=0,view=0;
                                //2.解析init
                                String[] aggArray = init.split(",");//["COLLECTION:0","COMMENT:0",""]
                                for (String agg : aggArray) {//agg=COLLECTION:0
                                    String[] split = agg.split(":");//split[0]= COLLECTION  split[1]=0
                                    //表示对比类型
                                    switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                        case COLLECTION:
                                            collection+=Integer.parseInt(split[1]);
                                            break;
                                        case COMMENT:
                                            comment+=Integer.parseInt(split[1]);
                                            break;
                                        case LIKES:
                                            like+=Integer.parseInt(split[1]);
                                            break;
                                        case VIEWS:
                                            view+=Integer.parseInt(split[1]);
                                            break;
                                    }
                                }
                                //3.解析value  value=COLLECTION:-1
                                String[] valueSplit = value.split(":");//split[0]= COLLECTION  split[1]=-1
                                //表示对比类型
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(valueSplit[0])){
                                    case COLLECTION:
                                        collection+=Integer.parseInt(valueSplit[1]);
                                        break;
                                    case COMMENT:
                                        comment+=Integer.parseInt(valueSplit[1]);
                                        break;
                                    case LIKES:
                                        like+=Integer.parseInt(valueSplit[1]);
                                        break;
                                    case VIEWS:
                                        view+=Integer.parseInt(valueSplit[1]);
                                        break;
                                }

                                //4.返回聚合之后的数据，返回值必须是字符串
                                String formatStr=String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d",collection,comment,like,view);
                                System.err.println(formatStr);
                                return formatStr;
                            }
                        },
                        /**
                         * 给此次聚合取一个名称
                         */
                        Materialized.as("hot-article-name"))
                .toStream()//转化类型为Kstream
        /**
         * 把聚合处理之后的数据进行封装，然后发送给kafka
         * 此时的key表示什么？  文章id，也叫消息内容key
         * 此时的value表示什么？ 表示聚合之后的返回值字符串“COLLECTION:5,COMMENT:1,LIKES:1,VIEWS:2”  消息内容
         */
                .map((key,value)->{
                    return new KeyValue<String,String >(key.key().toString(),formatMess(key.key().toString(),value.toString()));
                })
                //3.发送消息
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        //4.返回数据
        return kStream;
    }

    /**
     * 抽取的方法，主要是用来处理发送消息内容
     * @param key  文章id
     * @param value  "COLLECTION:5,COMMENT:1,LIKES:1,VIEWS:2"
     * @return  必须包含文章id和value值
     */
    private String formatMess(String key, String value) {
        ArticleVisitStreamMess mess=new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));

        String[] aggArray = value.split(",");//["COLLECTION:0","COMMENT:0",""]
        for (String agg : aggArray) {//agg=COLLECTION:0
            String[] split = agg.split(":");//split[0]= COLLECTION  split[1]=0
            //表示对比类型
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COLLECTION:
                   mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        return JSON.toJSONString(mess);
    }
}
